package com.iot.socket;

import com.iot.compont.ServerNetwork;
import com.iot.gateway.GatewayEntity;
import com.iot.messaage.FromDeviceMessageContext;
import com.iot.network.WebSocketDeviceSession;
import com.iot.test.TestDeviceMessageDecoder;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.net.SocketAddress;
import reactor.core.Disposable;
import reactor.core.scheduler.Schedulers;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 */
public class SoketServerNetwork implements ServerNetwork {
    private Vertx vertx;
    private Collection<HttpServer> instances;
    public GatewayEntity gatewayEntity;
    private Disposable disposable;
//    String networkType = GatewayServerEnum.SOCKET.name();
    @Override
    public SoketServerNetwork createServer(GatewayEntity gatewayEntity) {
        this.gatewayEntity = gatewayEntity;
        vertx = Vertx.vertx();
        int instance = Math.max(2, Runtime.getRuntime().availableProcessors());
        instances = new ArrayList<>(instance);
        SocketAddress socketAddress = SocketAddress.inetSocketAddress(gatewayEntity.getProd(), gatewayEntity.getHost());
        for (int i = 0; i < instance; i++) {
            instances.add(vertx.createHttpServer());
        }
        for (HttpServer socketService : instances) {
            {
                socketService.webSocketHandler(this::handleConnection);
                vertx.nettyEventLoopGroup().execute(() -> socketService.
                        listen(socketAddress, result -> {
                            if (result.succeeded()) {

                            } else {
                                System.out.println("socket-io失败");
                            }
                        }));
                socketService.exceptionHandler(err -> {
                    System.out.println(err.getMessage());
                });
            }
        }
        return this;
    }

    @Override
    public Runnable cloneServer() {
        return () -> {
            for (HttpServer httpServer : instances) {
                httpServer.close(httpServerAsyncResult -> {
                    if (httpServerAsyncResult.succeeded()) {
                    } else {

                    }
                });
            }
            instances.clear(); // 清除列表
        };
    }

    public void handleConnection(ServerWebSocket serverWebSocket) {
        String uri = serverWebSocket.uri();
        serverWebSocket.handler(buffer -> {
            System.out.println("Received data: " + buffer.toString());

            WebSocketDeviceSession session=new WebSocketDeviceSession(serverWebSocket);
            TestDeviceMessageDecoder testDeviceMessageDecoder=new TestDeviceMessageDecoder();
            testDeviceMessageDecoder.decode(FromDeviceMessageContext.of(session));

            disposable.dispose();
            disposable = Schedulers
                    .parallel()
                    .schedule(this.closeSocket(serverWebSocket), 10, TimeUnit.SECONDS);
        });
        disposable = Schedulers
                .parallel()
                .schedule(this.closeSocket(serverWebSocket), 10, TimeUnit.SECONDS);
    }

    public Runnable closeSocket(ServerWebSocket serverWebSocket) {
        return () -> {
            serverWebSocket.close();
        };
    }
}
